package com.hex.es.bulk;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.hex.es.autoconfig.EsParamProperties;
import com.hex.es.entity.EsInfoDomain;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * es信息处理控制中心
 *
 * @author hex
 * @date 2019/11/01
 */
@Slf4j
@Component
public class EsOperationControlCenter implements ApplicationRunner {

    /**队列最大容忍数量，建议不要太大，队列用到size操作，时间复杂度为O(N)*/
    private static volatile int THRESHOLD = 1000;

    /**间隔多长时间执行一次，单位：秒*/
    private static volatile int INTERVAL = 10;

    /**自增控制变量*/
    private static volatile AtomicInteger VARIABLE = new AtomicInteger(0);

    private ExecutorService pool;

    @Resource
    private EsParamProperties esParamProperties;

    @Resource
    private EsOperationLogicRunnable operationLogicRunnable;

    @Override
    public void run(ApplicationArguments args) {
        if(esParamProperties.getTaskSwitch()) {
            ThreadFactory namedThreadFactory = new ThreadFactoryBuilder()
                    .setNameFormat("es-handle-%d").build();
            ScheduledThreadPoolExecutor exec = new ScheduledThreadPoolExecutor(1, namedThreadFactory);
            exec.scheduleAtFixedRate(this::controlPolling, 5, 1, TimeUnit.SECONDS);
        }

        ThreadFactory threadFactory = new ThreadFactoryBuilder()
                .setNameFormat("kafka-to-es-%d").build();
        pool = Executors. newSingleThreadExecutor(threadFactory);
    }

    /**
     * 添加es操作数据
     * 入口
     * @param esInfoDomain
     */
    public void addEvent(EsInfoDomain esInfoDomain){
        EsInfoStorageCache.add(esInfoDomain);
        notifyControl();
    }

    private void controlPolling(){
        if (VARIABLE.get() >= INTERVAL) {
            refresh();
            operationLogicRunnable.run();
        } else {
            VARIABLE.incrementAndGet();
        }
    }

    /**
     * 通知控制器处理
     */
    private void notifyControl(){
        if (EsInfoStorageCache.size() >= THRESHOLD) {
            log.info("添加es操作数据已达到阈值，开始执行es批量操作");
            refresh();
            pool.submit(() -> operationLogicRunnable.run());
            log.info("批量执行完成");
        }
    }

    private void refresh() {
        VARIABLE.set(0);
    }

    @PreDestroy
    public void destroy(){
        operationLogicRunnable.run();
    }

    public static void setThreshold(int threshold) {
        EsOperationControlCenter.THRESHOLD = threshold;
    }

    public static int getThreshold() {
        return THRESHOLD;
    }

    public static int getInterval() {
        return INTERVAL;
    }

    public static void setInterval(int interval) {
        EsOperationControlCenter.INTERVAL = interval;
    }

}
